package com.atguigu.chapter5.source;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;


import java.io.*;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**
 * @ClassName: Clustom_Source
 * @Description:
 * @Author: kele
 * @Date: 2021/4/1 14:50
 **/

/**
 * 使用自定义的source
 */
public class Clustom_Source {


    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        DataStreamSource<WaterSensor> ds = env.addSource(new mysource("hadoop162", 8888));
        ds.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 声明一个静态内部类
     */
    public static class mysource implements SourceFunction<WaterSensor> {

        private String host;
        private Integer port;

        private Socket socket;
        private InputStream inputStream;

        private Boolean flag = true;
        private InputStreamReader inputStreamReader;
        private BufferedReader bufferedReader;

        public mysource(String host,Integer port){
            this.host = host;
            this.port = port;
        }

        @Override
        public void run(SourceContext ctx) throws Exception {


            socket = new Socket(host, port);

            //1、从socket中获取字节流
            inputStream= socket.getInputStream();
            //2、将字节流转化为字符流
            inputStreamReader = new InputStreamReader(inputStream, StandardCharsets.UTF_8);

            //3、按行读取数据，转化为buffer
            bufferedReader = new BufferedReader(inputStreamReader);

            String line = bufferedReader.readLine();

            while (line != null && flag){

                String[] data = line.split(",");

                if(data.length == 3){

                    WaterSensor waterSensor = new WaterSensor(data[0],Long.valueOf(data[1]),Integer.parseInt(data[2]));
                    ctx.collect(waterSensor);
                }

                line= bufferedReader.readLine();
            }


        }

        @Override
        public void cancel() {
            flag = false;

            if(socket != null){
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }


            if (inputStream != null) {
                try {
                    inputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }


            if (bufferedReader != null) {

                try {
                    bufferedReader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }
    }
}


